package topN;

import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

public class TotalPriceReducer extends Reducer<DoubleWritable, Text, DoubleWritable, Text> {

    private static final int TOP_N = 10;
    private List<Pair<Double, String>> priceList = new ArrayList<>();

    @Override
    protected void reduce(DoubleWritable key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
        for (Text value : values) {
            priceList.add(new Pair<>(key.get(), value.toString()));
        }
    }

    @Override
    protected void cleanup(Context context) throws IOException, InterruptedException {
        Collections.sort(priceList, (o1, o2) -> Double.compare(o2.getKey(), o1.getKey()));

        int count = 0;
        for (Pair<Double, String> pair : priceList) {
            if (count >= TOP_N) break;
            context.write(new DoubleWritable(pair.getKey()), new Text(pair.getValue()));
            count++;
        }
    }

    private static class Pair<K extends Comparable<? super K>, V> implements Comparable<Pair<K, V>> {
        private K key;
        private V value;

        public Pair(K key, V value) {
            this.key = key;
            this.value = value;
        }

        public K getKey() {
            return key;
        }

        public V getValue() {
            return value;
        }

        @Override
        public int compareTo(Pair<K, V> other) {
            return this.key.compareTo(other.key);
        }
    }
}
